package com.seari.main.line17;



import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.seari.bean.iec104.AiData;
import com.seari.bean.iec104.DiData;
import com.seari.bean.iec104.SimpleMetric;
import com.seari.utils.JedisPoolUtils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

public class Line17MasterWorker implements Runnable
{
    //private final static String MQ_HOST = "101.68.81.146";
	public static String MQ_HOST;
	public static int MQ_PORT;
	public static String MQ_VIRTUALHOST;
	public static String MQ_USER;
	public static String MQ_PASS;
	public static String MQ_EXCHANGE;
	public static String MQ_ROUTINGKEY;
	private static ObjectMapper mapper = new ObjectMapper();
	private Jedis jedis = null;
    private final static int LINE_17 = 17;
    private Channel channel;
    final static Logger logger = LoggerFactory.getLogger(Line17MasterWorker.class);
	BlockingQueue<List<String>> aiQueue;
	BlockingQueue<List<String>> diQueue;
	
	public Line17MasterWorker(BlockingQueue<List<String>> aiQueue,BlockingQueue<List<String>> diQueue)
	{
		this.aiQueue = aiQueue;
		this.diQueue = diQueue;
	}

	@Override
	public void run()
	{
		
		ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(MQ_HOST);
        factory.setPort(MQ_PORT);
        factory.setVirtualHost(MQ_VIRTUALHOST);
        factory.setUsername(MQ_USER);
        factory.setPassword(MQ_PASS);
        Connection connection = null;
        try {
        		jedis = JedisPoolUtils.getJedis();
        		connection = factory.newConnection();
            	channel = connection.createChannel();
                process();
        } catch (Exception ex) {
        	Line17NewStarter.workerCount.decrementAndGet();
            logger.warn(ex.getMessage());
        }finally {
        	try
			{
				if(null != channel)
				{
					channel.close();
				}
				
				if(null != connection)
				{
					connection.close();
				}
			} catch (IOException e)
			{
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (TimeoutException e)
			{
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			jedis.close();
		}
		
	}
	
	public void process() throws Exception
	{
		while (true)
		{
			List<String> diDatas = diQueue.poll(3, TimeUnit.SECONDS);
			if(null == diDatas)
			{
				List<String> aiDatas = aiQueue.poll(3, TimeUnit.SECONDS);
				if(null == aiDatas)
				{
					logger.info("diQueue and aiQueue is null,sleep 3 seconds");
					TimeUnit.SECONDS.sleep(5);
					continue;
				}else {
					sendAiData(aiDatas);
				}
			}else {
				sendDiData(diDatas);
			}
			
		}
	}
	
	public void sendAiData(List<String> aiDatas) throws Exception
	{
		Pipeline pipeline = jedis.pipelined();
		for (String aiKey : aiDatas)
		{
			pipeline.hget("line17_AI", aiKey);
		}
		List<Object> results = pipeline.syncAndReturnAll();
		List<SimpleMetric> simpleMetrics = new ArrayList<>();
		for(int i = 0; i < results.size(); i++)
		{
			AiData aiData = Line17NewStarter.aiDataMap.get(aiDatas.get(i));
			if(aiData == null)
			{
				//logger.warn("illegal aiData :" + aiDatas.get(i));
				continue;
			}
			SimpleMetric simpleMetric = new SimpleMetric();
			simpleMetric.setType("YC");
			simpleMetric.setCompany("seisys");
			simpleMetric.setDevice(aiData.getDesc());
			simpleMetric.setIoa(aiData.getPoint());
			simpleMetric.setLine(LINE_17);
			simpleMetric.setSystem(aiData.getSystem());
			simpleMetric.setStation(aiData.getStationEn());
			simpleMetric.setTimestampNs(System.currentTimeMillis() * 1000);
			simpleMetric.setValue((float)(Math.round(aiData.getRatio() * Float.parseFloat((String) results.get(i))*100))/100);
			//期安修改后协议均传float值，无系数
			//simpleMetric.setValue((float)(Math.round(Float.parseFloat((String) results.get(i))*100))/100);
			simpleMetrics.add(simpleMetric);
		}
		if(simpleMetrics.size() < 1)
		{
			return;
		}
		String jsonStr = mapper.writeValueAsString(simpleMetrics);
		logger.info("send yc data ");
		logger.info("current ai queue size is :" + aiQueue.size());
		channel.basicPublish(MQ_EXCHANGE, MQ_ROUTINGKEY, new AMQP.BasicProperties.Builder().contentType("application/json").contentEncoding("UTF-8").deliveryMode(1).build(), jsonStr.getBytes("UTF-8"));
	}
	
	public void sendDiData(List<String> diDatas) throws Exception
	{
		Pipeline pipeline = jedis.pipelined();
		for (String diKey : diDatas)
		{
			pipeline.hget("line17_DI", diKey);
		}
		List<Object> results = pipeline.syncAndReturnAll();
		List<SimpleMetric> simpleMetrics = new ArrayList<>();
		for(int i = 0; i < results.size(); i++)
		{
			DiData diData = Line17NewStarter.diDataMap.get(diDatas.get(i));
			if(diData == null)
			{
				//logger.warn("illegal aiData :" + aiDatas.get(i));
				continue;
			}
			SimpleMetric simpleMetric = new SimpleMetric();
			simpleMetric.setType("YX");
			simpleMetric.setCompany("seisys");
			simpleMetric.setDevice(diData.getDesc());
			simpleMetric.setIoa(diData.getPoint());
			simpleMetric.setLine(LINE_17);
			simpleMetric.setSystem(diData.getSystem());
			simpleMetric.setStation(diData.getStationEn());
			simpleMetric.setTimestampNs(System.currentTimeMillis() * 1000);
			int result = Integer.parseInt((String) results.get(i));
			setDiValue(diData, simpleMetric,result);
			simpleMetrics.add(simpleMetric);
			//判断是否为双点遥信
			if(diData.getPoint2() != null && diData.getPoint2() > 0)
			{
				try
				{
					int point2Result = Integer.parseInt(jedis.hget("line17_DI", diData.getPoint2()+""));
					SimpleMetric simpleMetric2 = new SimpleMetric();
					simpleMetric2.setType("YX");
					simpleMetric2.setCompany("seisys");
					simpleMetric2.setDevice(diData.getDesc());
					simpleMetric2.setIoa(diData.getPoint2());
					simpleMetric2.setLine(LINE_17);
					simpleMetric2.setSystem(diData.getSystem());
					simpleMetric2.setStation(diData.getStationEn());
					simpleMetric2.setTimestampNs(System.currentTimeMillis() * 1000);
					setDiValue(diData, simpleMetric2,point2Result);
					simpleMetrics.add(simpleMetric2);
				} catch (Exception e)
				{
					// TODO: handle exception
					e.printStackTrace();
				}
			}
		}
		if(simpleMetrics.size() < 1)
		{
			return;
		}
		String jsonStr = mapper.writeValueAsString(simpleMetrics);
		logger.info("send yx data ");
		logger.info("current di queue size is :" + diQueue.size());
		channel.basicPublish(MQ_EXCHANGE, MQ_ROUTINGKEY, new AMQP.BasicProperties.Builder().contentType("application/json").contentEncoding("UTF-8").deliveryMode(1).build(), jsonStr.getBytes("UTF-8"));
	}
	
	private void setDiValue(DiData diData,SimpleMetric simpleMetric,int result)
	{
		if(diData.getReverse() == 1)
		{
			if(result == 1)
			{
				simpleMetric.setValue(0f);
			}else {
				simpleMetric.setValue(1f);
			}
		}else {
			if(result == 1)
			{
				simpleMetric.setValue(1f);
			}else {
				simpleMetric.setValue(0f);
			}
		}
	}
}
